package com.hww.mq.producer;

import com.hww.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Author: youdao.hww
 * @Date: 4/16/24
 * @Desc:
 */
@Slf4j
@Component
public class MQProducerService {

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendSimple(String topic, String tag, User user) {
        rocketMQTemplate.convertAndSend(topic + ":" + tag, user);
    }


    /**
     * Tag：用于区分过滤同一主题下的不同业务类型的消息，非常实用
     * 普通发送（这里的参数对象User可以随意定义，可以发送个对象，也可以是字符串等）
     */
    public void send(String topic, String tag, User user,String key) {
//        rocketMQTemplate.convertAndSend(topic + ":" + tag, user);
        rocketMQTemplate.send(topic + ":" + tag, MessageBuilder.withPayload(user).setHeader(RocketMQHeaders.KEYS,key).build()); // 等价于上面一行
    }


    /**
     * 带业务自定义的header发送消息
     */
    public void sendWithHeader(String topic, String tag, User user, Map<String, Object> maps) {
        rocketMQTemplate.convertAndSend(topic + ":" + tag, user, maps);
        rocketMQTemplate.send(topic + ":" + tag, MessageBuilder.withPayload(user).setHeader(RocketMQHeaders.KEYS, maps).build()); // 等价于上面一行
    }


    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String topic, String msgBody, int delayLevel) {
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public void sendOneWayMsg(String topic, String msgBody) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendTagMsg(String topic, String msgBody) {
        return rocketMQTemplate.syncSend(topic + ":tag2", MessageBuilder.withPayload(msgBody).build());
    }

    /***
     * 服务生产者，顺序消息
     * 把消息确保投递到同一条queue
     * 保证了消息的顺序性
     */
    public void sendFIFOMsg(String topic, String tag, User user, String headerKey) {
        //选择器规则构建
        rocketMQTemplate.setMessageQueueSelector((list, message, o) -> {
            int id = Integer.parseInt((String) o);
            int hash = (id % list.size());
            return list.get(hash);
        });
        Message<String> build = MessageBuilder.withPayload(user.toString()).setHeader(RocketMQHeaders.KEYS, headerKey).build();
        rocketMQTemplate.sendOneWayOrderly(topic + ":" + tag, build, String.valueOf(user.getId()));
    }


}
